package io.kiki.stack.netty.invoker;


import io.kiki.stack.netty.channelManager.Channel;
import io.kiki.stack.netty.channelManager.ChannelManager;
import io.kiki.stack.netty.channelManager.Url;
import io.kiki.stack.netty.command.*;
import io.kiki.stack.netty.config.switches.ProtocolSwitch;
import io.kiki.stack.netty.exception.RemotingException;
import io.kiki.stack.netty.exception.SerializationException;
import io.kiki.stack.netty.protocol.ProtocolImpl;
import io.kiki.stack.netty.util.AddressParser;
import io.kiki.stack.netty.util.InetSocketAddressUtil;
import io.kiki.stack.netty.util.TimerHolder;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.TimeUnit;

/**
 * Notice! DO NOT modify the request object concurrently when this method is called.
 */

@Slf4j
@Setter
@Getter
public abstract class Invoker {

    private final static long ABANDONING_REQUEST_THRESHOLD = 0L;
    public ChannelManager channelManager;
    protected CommandFactory commandFactory;


    public Invoker(CommandFactory commandFactory) {
        this.commandFactory = commandFactory;
    }


    public Invoker(CommandFactory commandFactory, ChannelManager channelManager) {
        this(commandFactory);

        this.channelManager = channelManager;
    }

    protected CommandFactory getCommandFactory() {
        if (this.commandFactory != null) {
            this.commandFactory = commandFactory;
        }
        CommandFactory commandFactory1 = ProtocolImpl.protocol.getCommandFactory();
        this.commandFactory = commandFactory1;
        return commandFactory1;
    }

    protected Command invokeSync(final Channel channel, final Command request, final int timeoutMillis) throws RemotingException, InterruptedException {
        int remainingTime = remainingTime(request, timeoutMillis);
        if (remainingTime <= ABANDONING_REQUEST_THRESHOLD) {
            // already timeout
            log.warn("already timeout before writing to the network, requestId: {}, remoting address: {}", request.getId(), channel.getUrl() != null ? channel.getUrl() : InetSocketAddressUtil.parseRemoteAddress(channel.getChannel()));
            return this.getCommandFactory().createTimeoutResponse(channel.getRemoteAddress());
        }

        final Future future = createInvokeFuture(request, request.getRequestContext());
        channel.addFuture(future);
        final int requestId = request.getId();
        RequestContext requestContext = request.getRequestContext();
        if (null != requestContext) {
            requestContext.put(RequestContext._PROCESS_CLIENT_BEFORE_SEND, System.nanoTime());
        }
        try {
            channel.getChannel().writeAndFlush(request).addListener(new ChannelFutureListener() {

                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if (!channelFuture.isSuccess()) {
                        channel.removeFuture(requestId);
                        future.putResponse(getCommandFactory().createSendFailedResponse(channel.getRemoteAddress(), channelFuture.cause()));
                        log.error("Invoke send failed, id={}", requestId, channelFuture.cause());
                    }
                }

            });
            if (null != requestContext) {
                requestContext.put(RequestContext._PROCESS_CLIENT_AFTER_SEND, System.nanoTime());
            }
        } catch (Exception e) {
            channel.removeFuture(requestId);
            future.putResponse(getCommandFactory().createSendFailedResponse(channel.getRemoteAddress(), e));
            log.error("Exception caught when sending invocation, id={}", requestId, e);
        }
        Command response = future.waitResponse(remainingTime);

        if (null != requestContext) {
            requestContext.put(RequestContext._PROCESS_CLIENT_RECEIVED, System.nanoTime());
        }

        if (response == null) {
            channel.removeFuture(requestId);
            response = this.getCommandFactory().createTimeoutResponse(channel.getRemoteAddress());
            log.warn("Wait response, request id={} timeout!", requestId);
        }

        return response;
    }

    protected void invokeWithCallback(final Channel channel, final Command request, final Callback callback, final int timeoutMillis) {
        final Future future = createInvokeFuture(channel, request, request.getRequestContext(), callback);
        int remainingTime = remainingTime(request, timeoutMillis);
        if (remainingTime <= ABANDONING_REQUEST_THRESHOLD) {
            log.warn("already timeout before writing to the network, requestId: {}, remoting address: {}", request.getId(), channel.getUrl() != null ? channel.getUrl() : InetSocketAddressUtil.parseRemoteAddress(channel.getChannel()));
            future.putResponse(getCommandFactory().createTimeoutResponse(channel.getRemoteAddress()));
            future.tryAsyncExecuteInvokeCallbackAbnormally();
            return;
        }
        channel.addFuture(future);
        final int requestId = request.getId();
        try {
            Timeout timeout = TimerHolder.getTimer().newTimeout(new TimerTask() {
                @Override
                public void run(Timeout timeout) throws Exception {
                    Future future = channel.removeFuture(requestId);
                    if (future != null) {
                        future.putResponse(getCommandFactory().createTimeoutResponse(channel.getRemoteAddress()));
                        future.tryAsyncExecuteInvokeCallbackAbnormally();
                    }
                }

            }, remainingTime, TimeUnit.MILLISECONDS);
            future.addTimeout(timeout);
            channel.getChannel().writeAndFlush(request).addListener(new ChannelFutureListener() {

                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if (!channelFuture.isSuccess()) {
                        Future future1 = channel.removeFuture(requestId);
                        if (future1 != null) {
                            future1.cancelTimeout();
                            future1.putResponse(getCommandFactory().createSendFailedResponse(channel.getRemoteAddress(), channelFuture.cause()));
                            future1.tryAsyncExecuteInvokeCallbackAbnormally();
                        }
                        log.error("Invoke send failed. The address is {}", InetSocketAddressUtil.parseRemoteAddress(channel.getChannel()), channelFuture.cause());
                    }
                }

            });
        } catch (Exception e) {
            Future f = channel.removeFuture(requestId);
            if (f != null) {
                f.cancelTimeout();
                f.putResponse(getCommandFactory().createSendFailedResponse(channel.getRemoteAddress(), e));
                f.tryAsyncExecuteInvokeCallbackAbnormally();
            }
            log.error("Exception caught when sending invocation. The address is {}", InetSocketAddressUtil.parseRemoteAddress(channel.getChannel()), e);
        }
    }

    protected Future invokeWithFuture(final Channel channel, final Command request, final int timeoutMillis) {

        final Future future = createInvokeFuture(request, request.getRequestContext());
        channel.addFuture(future);
        int remainingTime = remainingTime(request, timeoutMillis);
        if (remainingTime <= ABANDONING_REQUEST_THRESHOLD) {
            log.warn("already timeout before writing to the network, requestId: {}, remoting address: {}", request.getId(), channel.getUrl() != null ? channel.getUrl() : InetSocketAddressUtil.parseRemoteAddress(channel.getChannel()));
            future.putResponse(getCommandFactory().createTimeoutResponse(channel.getRemoteAddress()));
            return future;
        }

        final int requestId = request.getId();
        try {
            Timeout timeout = TimerHolder.getTimer().newTimeout(new TimerTask() {
                @Override
                public void run(Timeout timeout) throws Exception {
                    Future future = channel.removeFuture(requestId);
                    if (future != null) {
                        future.putResponse(getCommandFactory().createTimeoutResponse(channel.getRemoteAddress()));
                    }
                }

            }, remainingTime, TimeUnit.MILLISECONDS);
            future.addTimeout(timeout);

            channel.getChannel().writeAndFlush(request).addListener(new ChannelFutureListener() {

                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if (!channelFuture.isSuccess()) {
                        Future f = channel.removeFuture(requestId);
                        if (f != null) {
                            f.cancelTimeout();
                            f.putResponse(getCommandFactory().createSendFailedResponse(channel.getRemoteAddress(), channelFuture.cause()));
                        }
                        log.error("Invoke send failed. The address is {}", InetSocketAddressUtil.parseRemoteAddress(channel.getChannel()), channelFuture.cause());
                    }
                }

            });
        } catch (Exception e) {
            Future future1 = channel.removeFuture(requestId);
            if (future1 != null) {
                future1.cancelTimeout();
                future1.putResponse(getCommandFactory().createSendFailedResponse(channel.getRemoteAddress(), e));
            }
            log.error("Exception caught when sending invocation. The address is {}", InetSocketAddressUtil.parseRemoteAddress(channel.getChannel()), e);
        }
        return future;
    }

    protected void oneway(final Channel channel, final Command request) {
        if (channel == null) {
            log.error("conn is null");
            return;
        }

        Url url = channel.getUrl();
        if (url != null) {
            int remainingTime = remainingTime(request, url.getConnectTimeout());
            if (remainingTime <= ABANDONING_REQUEST_THRESHOLD) {
                log.warn("already timeout before writing to the network, requestId: {}, remoting address: {}", request.getId(), channel.getUrl() != null ? channel.getUrl() : InetSocketAddressUtil.parseRemoteAddress(channel.getChannel()));
                return;
            }
        }

        try {
            channel.getChannel().writeAndFlush(request).addListener(new ChannelFutureListener() {

                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if (!channelFuture.isSuccess()) {
                        log.error("Invoke send failed. The address is {}", InetSocketAddressUtil.parseRemoteAddress(channel.getChannel()), channelFuture.cause());
                    }
                }

            });
        } catch (Exception e) {
            log.error("Exception caught when sending invocation. The address is {}", InetSocketAddressUtil.parseRemoteAddress(channel.getChannel()), e);
        }
    }


    private int remainingTime(Command request, int timeout) {
        RequestContext requestContext = request.getRequestContext();
        if (requestContext == null) {
            return timeout;
        }
        Long cost = requestContext.get(RequestContext.CLIENT_CONN_CREATETIME);
        if (cost == null) {
            return timeout;
        }

        return (int) (timeout - cost);
    }

    public void oneway(final String addr, final Object request, final RequestContext requestContext) throws RemotingException, InterruptedException {
        Url url = AddressParser.parse(addr);
        this.oneway(url, request, requestContext);
    }

    public abstract void oneway(final Url url, final Object request, final RequestContext requestContext) throws RemotingException, InterruptedException;

    public void oneway(final Channel channel, final Object request, final RequestContext requestContext) throws RemotingException {
        RequestCommand requestCommand = (RequestCommand) toRemotingCommand(request, requestContext, -1);
        requestCommand.setType(RpcCommandType.REQUEST_ONEWAY);
        preProcessInvokeContext(requestContext, requestCommand, channel);
        oneway(channel, requestCommand);
    }

    public Object invokeSync(final String addr, final Object request, final RequestContext requestContext, final int timeoutMillis) throws RemotingException, InterruptedException {
        Url url = AddressParser.parse(addr);
        return this.invokeSync(url, request, requestContext, timeoutMillis);
    }

    public abstract Object invokeSync(final Url url, final Object request, final RequestContext requestContext, final int timeoutMillis) throws RemotingException, InterruptedException;

    public Object invokeSync(final Channel channel, final Object request, final RequestContext requestContext, final int timeoutMillis) throws RemotingException, InterruptedException {
        Command requestCommand = toRemotingCommand(request, requestContext, timeoutMillis);
        preProcessInvokeContext(requestContext, requestCommand, channel);
        ResponseCommand responseCommand = (ResponseCommand) invokeSync(channel, requestCommand, timeoutMillis);
        responseCommand.setRequestContext(requestContext);

        return RpcResponseResolver.resolveResponseObject(responseCommand, InetSocketAddressUtil.parseRemoteAddress(channel.getChannel()));
    }

    public RpcResponseFuture invokeWithFuture(final String addr, final Object request, final RequestContext requestContext, int timeoutMillis) throws RemotingException, InterruptedException {
        Url url = AddressParser.parse(addr);
        return this.invokeWithFuture(url, request, requestContext, timeoutMillis);
    }

    public abstract RpcResponseFuture invokeWithFuture(final Url url, final Object request, final RequestContext requestContext, final int timeoutMillis) throws RemotingException, InterruptedException;

    public RpcResponseFuture invokeWithFuture(final Channel channel, final Object request, final RequestContext requestContext, final int timeoutMillis) throws RemotingException {

        Command requestCommand = toRemotingCommand(request, requestContext, timeoutMillis);

        preProcessInvokeContext(requestContext, requestCommand, channel);
        Future future = invokeWithFuture(channel, requestCommand, timeoutMillis);
        return new RpcResponseFuture(InetSocketAddressUtil.parseRemoteAddress(channel.getChannel()), future);
    }

    public void invokeWithCallback(String addr, Object request, final RequestContext requestContext, Callback callback, int timeoutMillis) throws RemotingException, InterruptedException {
        Url url = AddressParser.parse(addr);
        this.invokeWithCallback(url, request, requestContext, callback, timeoutMillis);
    }

    public abstract void invokeWithCallback(final Url url, final Object request, final RequestContext requestContext, final Callback callback, final int timeoutMillis) throws RemotingException, InterruptedException;

    public void invokeWithCallback(final Channel channel, final Object request, final RequestContext requestContext, final Callback callback, final int timeoutMillis) throws RemotingException {
        Command requestCommand = toRemotingCommand(request, requestContext, timeoutMillis);
        preProcessInvokeContext(requestContext, requestCommand, channel);
        invokeWithCallback(channel, requestCommand, callback, timeoutMillis);
    }

    protected Command toRemotingCommand(Object request, RequestContext requestContext, int timeoutMillis) throws SerializationException {
        AbstractRequestCommand command = this.getCommandFactory().createRequest(request);

        if (null != requestContext) {

            // enable crc by default, user can disable by set invoke context `false` for key `InvokeContext._CRC_SWITCH`
            Boolean crcSwitch = requestContext.get(RequestContext._CRC_SWITCH, ProtocolSwitch.CRC_SWITCH_DEFAULT_VALUE);
            if (crcSwitch == null || crcSwitch) {
                command.setProtocolSwitch(ProtocolSwitch.create(new int[]{ProtocolSwitch.CRC_SWITCH_INDEX}));
            }
        } else {
            // enable crc by default, if there is no invoke context.
            command.setProtocolSwitch(ProtocolSwitch.create(new int[]{ProtocolSwitch.CRC_SWITCH_INDEX}));
        }
        command.setTimeout(timeoutMillis);
        command.setRequestClass(request.getClass().getName());
        command.setRequestContext(requestContext);
        command.serialize();
        //logDebugInfo(command);
        return command;
    }


    protected Future createInvokeFuture(Command request, RequestContext requestContext) {
        return new FutureImpl(request.getId(), null, null, this.getCommandFactory(), requestContext);
    }


    protected Future createInvokeFuture(Channel channel, Command request, RequestContext requestContext, Callback callback) {
        return new FutureImpl(request.getId(), new RpcCallbackListener(InetSocketAddressUtil.parseRemoteAddress(channel.getChannel())), callback, this.getCommandFactory(), requestContext);
    }

    protected abstract void preProcessInvokeContext(RequestContext requestContext, Command command, Channel channel);

}
